// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/common/base/portability.h>
#include <pollux/common/base/simd_util.h>
#include <pollux/dwio/common/decoder_util.h>
#include <pollux/dwio/common/selective_column_reader.h>
#include <pollux/dwio/common/type_util.h>

namespace kumo::pollux::dwio::common {
    // structs for extractValues in ColumnVisitor.

    dwio::common::NoHook &noHook();

    // Represents values not being retained after filter evaluation.

    struct DropValues {
        static constexpr bool kSkipNulls = false;
        using HookType = dwio::common::NoHook;

        bool acceptsNulls() const {
            return true;
        }

        template<typename V>
        void addValue(vector_size_t /*rowIndex*/, V /*value*/) {
        }

        template<typename T>
        void addNull(vector_size_t /*rowIndex*/) {
        }

        HookType &hook() {
            return noHook();
        }
    };

    class ExtractToReader {
    public:
        using HookType = dwio::common::NoHook;
        static constexpr bool kSkipNulls = false;

        explicit ExtractToReader(SelectiveColumnReader *reader) : reader_(reader) {
        }

        bool acceptsNulls() const {
            return true;
        }

        template<typename T>
        void addNull(vector_size_t /*rowIndex*/) {
            reader_->template addNull<T>();
        }

        template<typename V>
        void addValue(vector_size_t /*rowIndex*/, V value) {
            reader_->addValue(value);
        }

        dwio::common::NoHook &hook() {
            return noHook();
        }

    private:
        SelectiveColumnReader *const reader_;
    };

    template<typename THook>
    class ExtractToHook {
    public:
        using HookType = THook;
        static constexpr bool kSkipNulls = THook::kSkipNulls;

        explicit ExtractToHook(ValueHook *hook)
            : hook_(*reinterpret_cast<THook *>(hook)) {
        }

        bool acceptsNulls() {
            return hook_.acceptsNulls();
        }

        template<typename T>
        void addNull(vector_size_t rowIndex) {
            hook_.addNull(rowIndex);
        }

        template<typename V>
        void addValue(vector_size_t rowIndex, V value) {
            hook_.addValueTyped(rowIndex, value);
        }

        auto &hook() {
            return hook_;
        }

    private:
        THook hook_;
    };

    class ExtractToGenericHook {
    public:
        using HookType = ValueHook;
        static constexpr bool kSkipNulls = false;

        explicit ExtractToGenericHook(ValueHook *hook) : hook_(hook) {
        }

        bool acceptsNulls() const {
            return hook_->acceptsNulls();
        }

        template<typename T>
        void addNull(vector_size_t rowIndex) {
            hook_->addNull(rowIndex);
        }

        template<typename V>
        void addValue(vector_size_t rowIndex, V value) {
            hook_->addValueTyped(rowIndex, value);
        }

        ValueHook &hook() {
            return *hook_;
        }

    private:
        ValueHook *hook_;
    };

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    class DictionaryColumnVisitor;

    template<typename TFilter, typename ExtractValues, bool isDense>
    class StringDictionaryColumnVisitor;

    // Template parameter for controlling filtering and action on a set of rows.
    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    class ColumnVisitor {
    public:
        using FilterType = TFilter;
        using Extract = ExtractValues;
        using HookType = typename Extract::HookType;
        using DataType = T;
        static constexpr bool dense = isDense;
        static constexpr bool kHasBulkPath = true;
        static constexpr bool kHasFilter =
                !std::is_same_v<FilterType, pollux::common::AlwaysTrue>;
        static constexpr bool kHasHook = !std::is_same_v<HookType, NoHook>;
        static constexpr bool kFilterOnly = std::is_same_v<Extract, DropValues>;

        ColumnVisitor(
            TFilter &filter,
            SelectiveColumnReader *reader,
            const RowSet &rows,
            ExtractValues values)
            : filter_(filter),
              reader_(reader),
              allowNulls_(!TFilter::deterministic || filter.testNull()),
              rows_(&rows[0]),
              numRows_(rows.size()),
              rowIndex_(0),
              values_(values) {
        }

        bool allowNulls() {
            if (ExtractValues::kSkipNulls && TFilter::deterministic) {
                return false;
            }
            return allowNulls_ && values_.acceptsNulls();
        }

        vector_size_t start() {
            return isDense ? 0 : rowAt(0);
        }

        // Tests for a null value and processes it. If the value is not
        // null, returns 0 and has no effect. If the value is null, advances
        // to the next non-null value in 'rows_'. Returns the number of
        // values (not including nulls) to skip to get to the next non-null.
        // If there is no next non-null in 'rows_', sets 'atEnd'. If 'atEnd'
        // is set and a non-zero skip is returned, the caller must perform
        // the skip before returning.
        MELON_ALWAYS_INLINE vector_size_t checkAndSkipNulls(
            const uint64_t *nulls,
            vector_size_t &current,
            bool &atEnd) {
            auto testRow = currentRow();
            // Check that the caller and the visitor are in sync about current row.
            POLLUX_DCHECK(current == testRow);
            uint32_t nullIndex = testRow >> 6;
            uint64_t nullWord = nulls[nullIndex];
            if (nullWord == bits::kNotNull64) {
                return 0;
            }
            uint8_t nullBit = testRow & 63;
            if ((nullWord & (1UL << nullBit))) {
                return 0;
            }
            // We have a null. We find the next non-null.
            if (++rowIndex_ >= numRows_) {
                atEnd = true;
                return 0;
            }
            auto rowOfNullWord = testRow - nullBit;
            if (isDense) {
                if (nullBit == 63) {
                    nullBit = 0;
                    rowOfNullWord += 64;
                    nullWord = nulls[++nullIndex];
                } else {
                    ++nullBit;
                    // set all the bits below the row to null.
                    nullWord &= ~pollux::bits::lowMask(nullBit);
                }
                for (;;) {
                    auto nextNonNull = count_trailing_zeros(nullWord);
                    if (rowOfNullWord + nextNonNull >= numRows_) {
                        // Nulls all the way to the end.
                        atEnd = true;
                        return 0;
                    }
                    if (nextNonNull < 64) {
                        POLLUX_CHECK_LE(rowIndex_, rowOfNullWord + nextNonNull);
                        rowIndex_ = rowOfNullWord + nextNonNull;
                        current = currentRow();
                        return 0;
                    }
                    rowOfNullWord += 64;
                    nullWord = nulls[++nullIndex];
                }
            } else {
                // Sparse row numbers. We find the first non-null and count
                // how many non-nulls on rows not in 'rows_' we skipped.
                int32_t toSkip = 0;
                nullWord &= ~pollux::bits::lowMask(nullBit);
                for (;;) {
                    testRow = currentRow();
                    while (testRow >= rowOfNullWord + 64) {
                        toSkip += __builtin_popcountll(nullWord);
                        nullWord = nulls[++nullIndex];
                        rowOfNullWord += 64;
                    }
                    // testRow is inside nullWord. See if non-null.
                    nullBit = testRow & 63;
                    if ((nullWord & (1UL << nullBit))) {
                        toSkip +=
                                __builtin_popcountll(nullWord & pollux::bits::lowMask(nullBit));
                        current = testRow;
                        return toSkip;
                    }
                    if (++rowIndex_ >= numRows_) {
                        // We end with a null. Add the non-nulls below the final null.
                        toSkip += __builtin_popcountll(
                            nullWord & pollux::bits::lowMask(testRow - rowOfNullWord));
                        atEnd = true;
                        return toSkip;
                    }
                }
            }
        }

        vector_size_t processNull(bool &atEnd) {
            vector_size_t previous = currentRow();
            if (filter_.testNull()) {
                filterPassedForNull();
            } else {
                filterFailed();
            }
            if (++rowIndex_ >= numRows_) {
                atEnd = true;
                return rowAt(numRows_ - 1) - previous;
            }
            if (TFilter::deterministic && isDense) {
                return 0;
            }
            return currentRow() - previous - 1;
        }

        // Check if a string value doesn't pass the filter based on length.
        // Return unset optional if length is not sufficient to determine
        // whether the value passes or not. In this case, the caller must
        // call "process" for the actual string.
        MELON_ALWAYS_INLINE std::optional<vector_size_t> processLength(
            int32_t length,
            bool &atEnd) {
            if (!TFilter::deterministic) {
                return std::nullopt;
            }

            if (filter_.testLength(length)) {
                return std::nullopt;
            }

            filterFailed();

            if (++rowIndex_ >= numRows_) {
                atEnd = true;
                return 0;
            }
            if (isDense) {
                return 0;
            }
            return currentRow() - rowAt(rowIndex_ - 1) - 1;
        }

        MELON_ALWAYS_INLINE vector_size_t process(T value, bool &atEnd) {
            if (!TFilter::deterministic) {
                const auto previous = currentRow();
                if (pollux::common::applyFilter(filter_, value)) {
                    filterPassed(value);
                } else {
                    filterFailed();
                }
                if (++rowIndex_ >= numRows_) {
                    atEnd = true;
                    return rowAt(numRows_ - 1) - previous;
                }
                return currentRow() - previous - 1;
            }

            // The filter passes or fails and we go to the next row if any.
            if (pollux::common::applyFilter(filter_, value)) {
                filterPassed(value);
            } else {
                filterFailed();
            }
            if (++rowIndex_ >= numRows_) {
                atEnd = true;
                return 0;
            }
            if (isDense) {
                return 0;
            }
            return currentRow() - rowAt(rowIndex_ - 1) - 1;
        }

        // Returns space for 'size' items of T for a scan to fill. The scan
        // calls addResults and related to mark which elements are part of
        // the result.
        inline T *mutableValues(int32_t size) {
            return reader_->mutableValues<T>(size);
        }

        SelectiveColumnReader &reader() const {
            return *reader_;
        }

        inline vector_size_t rowAt(vector_size_t index) const {
            if (isDense) {
                return index;
            }
            return rows_[index];
        }

        vector_size_t rowIndex() const {
            return rowIndex_;
        }

        void setRowIndex(vector_size_t index) {
            rowIndex_ = index;
        }

        void addRowIndex(vector_size_t size) {
            rowIndex_ += size;
        }

        bool atEnd() const {
            return rowIndex_ >= numRows_;
        }

        vector_size_t currentRow() const {
            if (isDense) {
                return rowIndex_;
            }
            return rows_[rowIndex_];
        }

        const vector_size_t *rows() const {
            return rows_;
        }

        vector_size_t numRows() const {
            return numRows_;
        }

        void filterPassed(T value) {
            addResult(value);
            if (!std::is_same_v<TFilter, pollux::common::AlwaysTrue>) {
                addOutputRow(currentRow());
            }
        }

        inline void filterPassedForNull() {
            addNull();
            if (!std::is_same_v<TFilter, pollux::common::AlwaysTrue>) {
                addOutputRow(currentRow());
            }
        }

        MELON_ALWAYS_INLINE void filterFailed();

        inline void addResult(T value);

        inline void addNull();

        inline void addOutputRow(vector_size_t row);

        TFilter &filter() {
            return filter_;
        }

        int32_t *outputRows(int32_t size) {
            return reader_->mutableOutputRows(size);
        }

        int32_t numValuesBias() const {
            return numValuesBias_;
        }

        void setNumValuesBias(int32_t bias) {
            numValuesBias_ = bias;
        }

        void setNumValues(int32_t size) {
            reader_->setNumValues(numValuesBias_ + size);
            if (!std::is_same_v<TFilter, pollux::common::AlwaysTrue>) {
                reader_->setNumRows(numValuesBias_ + size);
            }
        }

        void addNumValues(int size) {
            auto numValues = reader_->numValues() + size;
            reader_->setNumValues(numValues);
            if constexpr (kHasFilter) {
                reader_->setNumRows(numValues);
            }
        }

        ExtractValues extractValues() const {
            return values_;
        }

        HookType &hook() {
            return values_.hook();
        }

        T *rawValues(int32_t size) {
            return reader_->mutableValues<T>(size);
        }

        uint64_t *rawNulls(int32_t size) {
            return reader_->mutable_nulls(size);
        }

        void setHasNulls() {
            reader_->setHasNulls();
        }

        void setAllNull(int32_t numValues) {
            setNumValues(numValues);
            reader_->setAllNull();
        }

        auto &innerNonNullRows() {
            return reader_->innerNonNullRows();
        }

        auto &outerNonNullRows() {
            return reader_->outerNonNullRows();
        }

        raw_vector<vector_size_t> &rowsCopy() const {
            return reader_->scanState().rowsCopy;
        }

        DictionaryColumnVisitor<T, TFilter, ExtractValues, isDense>
        toDictionaryColumnVisitor();

        StringDictionaryColumnVisitor<TFilter, ExtractValues, isDense>
        toStringDictionaryColumnVisitor();

        // Use for replacing all rows with non-null rows for fast path with processRun
        // and processRle.
        void setRows(melon::Range<const int32_t *> newRows) {
            rows_ = newRows.data();
            numRows_ = newRows.size();
        }

    protected:
        TFilter &filter_;
        SelectiveColumnReader *reader_;
        const bool allowNulls_;
        const vector_size_t *rows_;
        vector_size_t numRows_;
        vector_size_t rowIndex_;
        int32_t numValuesBias_{0};
        ExtractValues values_;
    };

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    MELON_ALWAYS_INLINE void
    ColumnVisitor<T, TFilter, ExtractValues, isDense>::filterFailed() {
        const auto preceding = filter_.getPrecedingPositionsToFail();
        const auto succeeding = filter_.getSucceedingPositionsToFail();
        if (preceding) {
            reader_->dropResults(preceding);
        }
        if (succeeding) {
            rowIndex_ += succeeding;
        }
    }

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    inline void ColumnVisitor<T, TFilter, ExtractValues, isDense>::addResult(
        T value) {
        values_.addValue(rowIndex_ + numValuesBias_, value);
    }

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    inline void ColumnVisitor<T, TFilter, ExtractValues, isDense>::addNull() {
        values_.template addNull<T>(rowIndex_ + numValuesBias_);
    }

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    inline void ColumnVisitor<T, TFilter, ExtractValues, isDense>::addOutputRow(
        vector_size_t row) {
        reader_->addOutputRow(row);
    }

    enum FilterResult { kUnknown = 0x40, kSuccess = 0x80, kFailure = 0 };

    namespace detail {
        template<typename T, typename A>
        struct LoadIndices;

        template<typename A>
        struct LoadIndices<int32_t, A> {
            static ksimd::batch<int32_t, A> apply(const int32_t *values, const A &) {
                return ksimd::load_unaligned<A>(values);
            }
        };

        template<typename A>
        struct LoadIndices<float, A> {
            static ksimd::batch<int32_t, A> apply(const float *values, const A &) {
                return ksimd::load_unaligned<A>(reinterpret_cast<const int32_t *>(values));
            }
        };

        template<typename A>
        struct LoadIndices<int16_t, A> {
            static ksimd::batch<int32_t, A> apply(
                const int16_t *values,
                const ksimd::generic &) {
                constexpr int N = ksimd::batch<int32_t, A>::size;
                alignas(A::alignment()) int32_t tmp[N];
                for (int i = 0; i < N; ++i) {
                    tmp[i] = values[i];
                }
                return ksimd::load_aligned(tmp);
            }

#if KSIMD_WITH_AVX2
            static ksimd::batch<int32_t, A> apply(
                const int16_t *values,
                const ksimd::avx2 &) {
                return _mm256_cvtepi16_epi32(
                    _mm_loadu_si128(reinterpret_cast<const __m128i *>(values)));
            }
#endif
        };

        template<typename A>
        struct LoadIndices<int64_t, A> {
            static ksimd::batch<int32_t, A> apply(const int64_t *values, const A &arch) {
                return simd::gather<int32_t, int32_t, 8>(
                    reinterpret_cast<const int32_t *>(values),
                    simd::iota<int32_t>(arch),
                    arch);
            }
        };

        template<typename A>
        struct LoadIndices<int128_t, A> {
            static ksimd::batch<int32_t, A> apply(const int128_t *values, const A &arch) {
                POLLUX_UNREACHABLE();
            }
        };

        template<typename A>
        struct LoadIndices<double, A> {
            static ksimd::batch<int32_t, A> apply(const double *values, const A &arch) {
                return simd::gather<int32_t, int32_t, 8>(
                    reinterpret_cast<const int32_t *>(values),
                    simd::iota<int32_t>(arch),
                    arch);
            }
        };
    } // namespace detail

    template<typename T, typename A = ksimd::default_arch>
    inline ksimd::batch<int32_t> loadIndices(const T *values, const A &arch = {}) {
        return detail::LoadIndices<T, A>::apply(values, arch);
    }

    // Copies from 'input' to 'values' and translates  via 'dict'. Only elements
    // where 'dictMask' is true at the element's index are translated, else they are
    // passed as is. The elements of input that are copied to values with or without
    // translation are given by the first 'numBits' elements of 'selected'. There is
    // a generic and a V32 specialization of this template. The V32 specialization
    // has 'indices' holding the data to translate, which is loaded from input +
    // inputIndex.
    template<typename T>
    inline void storeTranslatePermute(
        const T *input,
        int32_t inputIndex,
        ksimd::batch<int32_t> /*indices*/,
        int selected,
        ksimd::batch_bool<int32_t> dictMask,
        int8_t numBits,
        const T *dict,
        T *values) {
        using TIndex = typename make_index<T>::type;
        auto selectedIndices = simd::byteSetBits(selected);
        auto inDict = simd::toBitMask(dictMask);
        for (auto i = 0; i < numBits; ++i) {
            if (inDict & (1 << selectedIndices[i])) {
                auto index = reinterpret_cast<const TIndex *>(
                    input)[inputIndex + selectedIndices[i]];
                if (sizeof(T) == 2) {
                    index &= 0xffff;
                }
                auto value = dict[index];
                values[i] = value;
            } else {
                auto value = input[inputIndex + selectedIndices[i]];
                values[i] = value;
            }
        }
    }

    template<>
    inline void storeTranslatePermute(
        const int32_t * /*input*/,
        int32_t /*inputIndex*/,
        ksimd::batch<int32_t> indices,
        int selected,
        ksimd::batch_bool<int32_t> dictMask,
        int8_t /*numBits*/,
        const int32_t *dict,
        int32_t *values) {
        auto translated = simd::maskGather(indices, dictMask, dict, indices);
        simd::filter(translated, selected).store_unaligned(values);
    }

    // Stores 8 elements starting at 'input' + 'inputIndex' into
    // 'values'. The values are translated via 'dict' for the positions
    // that are true in 'dictMask'.
    template<typename T>
    inline void storeTranslate(
        const T *input,
        int32_t inputIndex,
        ksimd::batch<int32_t> /*indices*/,
        ksimd::batch_bool<int32_t> dictMask,
        const T *dict,
        T *values) {
        using TIndex = typename make_index<T>::type;
        auto inDict = simd::toBitMask(dictMask);
        for (auto i = 0; i < dictMask.size; ++i) {
            if (inDict & (1 << i)) {
                auto index = reinterpret_cast<const TIndex *>(input)[inputIndex + i];
                values[i] = dict[index];
            } else {
                auto value = input[inputIndex + i];
                values[i] = value;
            }
        }
    }

    template<>
    inline void storeTranslate(
        const int32_t * /*input*/,
        int32_t /*inputIndex*/,
        ksimd::batch<int32_t> indices,
        ksimd::batch_bool<int32_t> dictMask,
        const int32_t *dict,
        int32_t *values) {
        simd::maskGather(indices, dictMask, dict, indices).store_unaligned(values);
    }

    namespace detail {
#if KSIMD_WITH_AVX2
        inline ksimd::batch<int64_t> cvtU32toI64(
            ksimd::batch<int32_t, ksimd::sse2> values) {
            return _mm256_cvtepu32_epi64(values);
        }
#elif KSIMD_WITH_SSE2 || KSIMD_WITH_NEON
inline ksimd::batch<int64_t> cvtU32toI64(simd::Batch64<int32_t> values) {
  int64_t lo = static_cast<uint32_t>(values.data[0]);
  int64_t hi = static_cast<uint32_t>(values.data[1]);
  return ksimd::batch<int64_t>(lo, hi);
}
#endif
    } // namespace detail

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    class DictionaryColumnVisitor
            : public ColumnVisitor<T, TFilter, ExtractValues, isDense> {
        using super = ColumnVisitor<T, TFilter, ExtractValues, isDense>;

    public:
        DictionaryColumnVisitor(
            TFilter &filter,
            SelectiveColumnReader *reader,
            const RowSet &rows,
            ExtractValues values)
            : ColumnVisitor<T, TFilter, ExtractValues, isDense>(
                  filter,
                  reader,
                  rows,
                  values),
              width_(
                  reader->fileType().type()->kind() == TypeKind::BIGINT
                      ? 8
                      : reader->fileType().type()->kind() == TypeKind::INTEGER
                            ? 4
                            : 2),
              state_(reader->scanState().rawState) {
        }

        MELON_ALWAYS_INLINE bool isInDict() {
            if (inDict()) {
                return bits::isBitSet(inDict(), super::currentRow());
            }
            return true;
        }

        MELON_ALWAYS_INLINE vector_size_t
        process(typename make_index<T>::type value, bool &atEnd) {
            if (!isInDict()) {
                // If reading fixed width values, the not in dictionary value will be read
                // as unsigned at the width of the type. Integer columns are signed, so
                // sign extend the value here.
                T signedValue;
                if (LIKELY(width_ == 8)) {
                    signedValue = value;
                } else if (width_ == 4) {
                    signedValue = static_cast<int32_t>(value);
                } else {
                    signedValue = static_cast<int16_t>(value);
                }
                return super::process(signedValue, atEnd);
            }

            const vector_size_t previous =
                    isDense && TFilter::deterministic ? 0 : super::currentRow();
            const T valueInDictionary = dict()[value];
            if constexpr (!hasFilter()) {
                super::filterPassed(valueInDictionary);
            } else {
                // check the dictionary cache
                if (TFilter::deterministic &&
                    filterCache()[value] == FilterResult::kSuccess) {
                    super::filterPassed(valueInDictionary);
                } else if (
                    TFilter::deterministic &&
                    filterCache()[value] == FilterResult::kFailure) {
                    super::filterFailed();
                } else {
                    if (pollux::common::applyFilter(super::filter_, valueInDictionary)) {
                        super::filterPassed(valueInDictionary);
                        if (TFilter::deterministic) {
                            filterCache()[value] = FilterResult::kSuccess;
                        }
                    } else {
                        super::filterFailed();
                        if (TFilter::deterministic) {
                            filterCache()[value] = FilterResult::kFailure;
                        }
                    }
                }
            }

            if (++super::rowIndex_ >= super::numRows_) {
                atEnd = true;
                return (isDense && TFilter::deterministic)
                           ? 0
                           : super::rowAt(super::numRows_ - 1) - previous;
            }
            if (isDense && TFilter::deterministic) {
                return 0;
            }
            return super::currentRow() - previous - 1;
        }

        // Processes 'numInput' dictionary indices in 'input'. Sets 'values'
        // and 'numValues'' to the resulting values. If hasFilter is true,
        // only values passing filter are put in 'values' and the indices of
        // the passing rows are put in the corresponding position in
        // 'filterHits'. 'scatterRows' may be non-null if there is no filter and the
        // decoded values should be scattered into values with gaps in between so as
        // to leave gaps  for nulls. If scatterRows is given, the ith value goes to
        // values[scatterRows[i]], else it goes to 'values[i]'. If 'hasFilter' is
        // true, the passing values are written to consecutive places in 'values'.
        template<bool hasFilter, bool hasHook, bool scatter>
        void processRun(
            const T *input,
            int32_t numInput,
            const int32_t *scatterRows,
            int32_t *filterHits,
            T *values,
            int32_t &numValues) {
            DKCHECK_EQ(input, values + numValues);
            if constexpr (!DictionaryColumnVisitor::hasFilter()) {
                if (hasHook) {
                    translateByDict(input, numInput, values);
                    super::values_.hook().addValues(
                        scatter
                            ? scatterRows + super::rowIndex_
                            : pollux::iota(
                                  super::numRows_,
                                  super::innerNonNullRows(),
                                  super::numValuesBias_) +
                              super::rowIndex_,
                        values,
                        numInput);
                    super::rowIndex_ += numInput;
                    return;
                }
                if constexpr (std::is_same_v<TFilter, pollux::common::IsNotNull>) {
                    auto *begin = (scatter ? scatterRows : super::rows_) + super::rowIndex_;
                    std::copy(begin, begin + numInput, filterHits + numValues);
                    if constexpr (!super::kFilterOnly) {
                        translateByDict(input, numInput, values + numValues);
                    }
                    numValues += numInput;
                } else {
                    if (inDict()) {
                        translateScatter<true, scatter>(
                            input, numInput, scatterRows, numValues, values);
                    } else {
                        translateScatter<false, scatter>(
                            input, numInput, scatterRows, numValues, values);
                    }
                    numValues = scatter
                                    ? scatterRows[super::rowIndex_ + numInput - 1] + 1
                                    : numValues + numInput;
                }
                super::rowIndex_ += numInput;
                return;
            } else {
                static_assert(hasFilter);
            }
            // The filter path optionally extracts values but always sets
            // filterHits. It first loads a vector of indices. It translates
            // those indices that refer to dictionary via the dictionary in
            // bulk. It checks the dictionary filter cache 8 values at a
            // time. It calls the scalar filter for the indices that were not
            // found in the cache. It gets a bitmask of up to 8 filter
            // results. It stores these in filterHits. If values are to be
            // written, the passing bitmap is used to load a permute mask to
            // permute the passing values to the left of a vector register and
            // write  the whole register to the end of 'values'
            constexpr bool kFilterOnly = super::kFilterOnly;
            constexpr int32_t kWidth = ksimd::batch<int32_t>::size;
            int32_t last = numInput & ~(kWidth - 1);
            for (auto i = 0; i < numInput; i += kWidth) {
                int8_t width = UNLIKELY(i == last) ? numInput - last : kWidth;
                auto indices = loadIndices(input + i);
                ksimd::batch_bool<int32_t> dictMask;
                if (inDict()) {
                    if (simd::isDense(super::rows_ + super::rowIndex_ + i, width)) {
                        dictMask = load8MaskDense(
                            inDict(), super::rows_[super::rowIndex_ + i], width);
                    } else {
                        dictMask = load8MaskSparse(
                            inDict(), super::rows_ + super::rowIndex_ + i, width);
                    }
                } else {
                    dictMask = simd::leadingMask<int32_t>(width);
                }

                // Load 8 filter cache values. Defaults the extra to values to 0 if
                // loading less than 8.
                auto cache = simd::maskGather<int32_t, int32_t, 1>(
                    ksimd::broadcast<int32_t>(0),
                    dictMask,
                    reinterpret_cast<const int32_t *>(filterCache() - 3),
                    indices);
                auto unknowns = simd::toBitMask(ksimd::batch_bool<int32_t>(
                    simd::reinterpretBatch<uint32_t>((cache & (kUnknown << 24)) << 1)));
                auto passed = simd::toBitMask(
                    ksimd::batch_bool<int32_t>(simd::reinterpretBatch<uint32_t>(cache)));
                if (UNLIKELY(unknowns)) {
                    uint16_t bits = unknowns;
                    // Ranges only over inputs that are in dictionary, the not in dictionary
                    // were masked off in 'dictMask'.
                    using TIndex = typename make_index<T>::type;
                    while (bits) {
                        int index = bits::getAndClearLastSetBit(bits);
                        auto value = reinterpret_cast<const TIndex *>(input)[i + index];
                        if (applyFilter(super::filter_, dict()[value])) {
                            filterCache()[value] = FilterResult::kSuccess;
                            passed |= 1 << index;
                        } else {
                            filterCache()[value] = FilterResult::kFailure;
                        }
                    }
                }
                // Were there values not in dictionary?
                if (inDict()) {
                    auto mask = simd::toBitMask(dictMask);
                    const auto allTrue = simd::allSetBitMask<int32_t>();
                    if (mask != allTrue) {
                        uint16_t bits = (allTrue ^ mask) & bits::lowMask(kWidth);
                        while (bits) {
                            auto index = bits::getAndClearLastSetBit(bits);
                            if (i + index >= numInput) {
                                break;
                            }
                            if (pollux::common::applyFilter(super::filter_, input[i + index])) {
                                passed |= 1 << index;
                            }
                        }
                    }
                }
                // We know 8 compare results. If all false, process next batch.
                if (!passed) {
                    continue;
                } else if (passed == (1 << ksimd::batch<int32_t>::size) - 1) {
                    // All passed, no need to shuffle the indices or values, write then to
                    // 'values' and 'filterHits'.
                    ksimd::load_unaligned(
                                (scatter ? scatterRows : super::rows_) + super::rowIndex_ + i)
                            .store_unaligned(filterHits + numValues);
                    if (!kFilterOnly) {
                        storeTranslate(
                            input, i, indices, dictMask, dict(), values + numValues);
                    }
                    numValues += kWidth;
                } else {
                    // Some passed. Permute  the passing row numbers and values to the left
                    // of the SIMD vector and store.
                    int8_t numBits = __builtin_popcount(passed);
                    simd::filter(
                                ksimd::load_unaligned(
                                    (scatter ? scatterRows : super::rows_) + super::rowIndex_ + i),
                                passed)
                            .store_unaligned(filterHits + numValues);
                    if (!kFilterOnly) {
                        storeTranslatePermute(
                            input,
                            i,
                            indices,
                            passed,
                            dictMask,
                            numBits,
                            dict(),
                            values + numValues);
                    }
                    numValues += numBits;
                }
            }
            super::rowIndex_ += numInput;
        }

        template<bool hasFilter, bool hasHook, bool scatter>
        void processRle(
            typename make_index<T>::type value,
            typename make_index<T>::type delta,
            int32_t numRows,
            int32_t currentRow,
            const int32_t *scatterRows,
            int32_t *filterHits,
            T *values,
            int32_t &numValues) {
            auto indices = reinterpret_cast<typename make_index<T>::type *>(values);
            if (sizeof(T) == 8) {
                constexpr int32_t kWidth = ksimd::batch<int64_t>::size;
                for (auto i = 0; i < numRows; i += kWidth) {
                    auto numbers = detail::cvtU32toI64(
                                       simd::loadGatherIndices<int64_t>(
                                           super::rows_ + super::rowIndex_ + i) -
                                       currentRow) *
                                   delta +
                                   value;
                    numbers.store_unaligned(indices + numValues + i);
                }
            } else if (sizeof(T) == 4) {
                constexpr int32_t kWidth = ksimd::batch<int32_t>::size;
                for (auto i = 0; i < numRows; i += kWidth) {
                    auto numbers =
                            (ksimd::load_unaligned(super::rows_ + super::rowIndex_ + i) -
                             currentRow) *
                            static_cast<int32_t>(delta) +
                            static_cast<int32_t>(value);
                    numbers.store_unaligned(indices + numValues + i);
                }
            } else {
                for (auto i = 0; i < numRows; ++i) {
                    indices[numValues + i] =
                            (super::rows_[super::rowIndex_ + i] - currentRow) * delta + value;
                }
            }

            processRun<hasFilter, hasHook, scatter>(
                values + numValues,
                numRows,
                scatterRows,
                filterHits,
                values,
                numValues);
        }

    private:
        template<bool hasInDict, bool scatter>
        void translateScatter(
            const T *input,
            int32_t numInput,
            const int32_t *scatterRows,
            int32_t numValues,
            T *values) {
            using TIndex = typename make_index<T>::type;

            for (int32_t i = numInput - 1; i >= 0; --i) {
                T value;
                if (hasInDict) {
                    if (bits::isBitSet(inDict(), super::rows_[super::rowIndex_ + i])) {
                        value = dict()[reinterpret_cast<const TIndex *>(input)[i]];
                    } else if (!scatter) {
                        continue;
                    } else {
                        value = input[i];
                    }
                } else {
                    value = dict()[reinterpret_cast<const TIndex *>(input)[i]];
                }
                if (scatter) {
                    values[scatterRows[super::rowIndex_ + i]] = value;
                } else {
                    values[numValues + i] = value;
                }
            }
        }

        // Returns 'numBits' bits starting at bit 'index' in 'bits' as a
        // 8x32 mask. This is used as a mask for maskGather to load selected
        // lanes from a dictionary.
        ksimd::batch_bool<int32_t>
        load8MaskDense(const uint64_t *bits, int32_t index, int8_t numBits) {
            uint8_t shift = index & 7;
            uint32_t byte = index >> 3;
            auto asBytes = reinterpret_cast<const uint8_t *>(bits);
            auto mask = (*reinterpret_cast<const int16_t *>(asBytes + byte) >> shift) &
                        bits::lowMask(numBits);
            return simd::fromBitMask<int32_t>(mask);
        }

        // Returns 'numBits' bits at bit offsets in 'rows' from 'bits' as a
        // 8x32 mask for use in maskGather.
        ksimd::batch_bool<int32_t>
        load8MaskSparse(const uint64_t *bits, const int32_t *rows, int8_t numRows) {
            return simd::fromBitMask<int32_t>(simd::gather8Bits(bits, rows, numRows));
        }

        void translateByDict(const T *values, int numValues, T *out) {
            using TIndex = typename make_index<T>::type;
            if (!inDict()) {
                for (auto i = 0; i < numValues; ++i) {
                    out[i] = dict()[reinterpret_cast<const TIndex *>(values)[i]];
                }
            } else if (super::dense) {
                bits::forEachSetBit(
                    inDict(),
                    super::rowIndex_,
                    super::rowIndex_ + numValues,
                    [&](int row) {
                        auto valueIndex = row - super::rowIndex_;
                        out[valueIndex] =
                                dict()[reinterpret_cast<const TIndex *>(values)[valueIndex]];
                        return true;
                    });
            } else {
                for (auto i = 0; i < numValues; ++i) {
                    if (bits::isBitSet(inDict(), super::rows_[super::rowIndex_ + i])) {
                        out[i] = dict()[reinterpret_cast<const TIndex *>(values)[i]];
                    }
                }
            }
        }

    protected:
        const uint64_t *inDict() const {
            return state_.inDictionary;
        }

        const T *dict() const {
            return reinterpret_cast<const T *>(state_.dictionary.values);
        }

        int32_t dictionarySize() const {
            return state_.dictionary.numValues;
        }

        uint8_t *filterCache() const {
            return state_.filterCache;
        }

        static constexpr bool hasFilter() {
            // Dictionary values cannot be null.  See the explanation in
            // `DictionaryValues::hasFilter'.
            return !std::is_same_v<TFilter, pollux::common::AlwaysTrue> &&
                   !std::is_same_v<TFilter, pollux::common::IsNotNull>;
        }

        const uint8_t width_;
        RawScanState state_;
    };

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    DictionaryColumnVisitor<T, TFilter, ExtractValues, isDense>
    ColumnVisitor<T, TFilter, ExtractValues, isDense>::toDictionaryColumnVisitor() {
        auto result = DictionaryColumnVisitor<T, TFilter, ExtractValues, isDense>(
            filter_, reader_, RowSet(rows_ + rowIndex_, numRows_), values_);
        result.numValuesBias_ = numValuesBias_;
        return result;
    }

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    StringDictionaryColumnVisitor<TFilter, ExtractValues, isDense>
    ColumnVisitor<T, TFilter, ExtractValues, isDense>::
    toStringDictionaryColumnVisitor() {
        auto result = StringDictionaryColumnVisitor<TFilter, ExtractValues, isDense>(
            filter_, reader_, RowSet(rows_ + rowIndex_, numRows_), values_);
        result.setNumValuesBias(numValuesBias_);
        return result;
    }

    template<typename TFilter, typename ExtractValues, bool isDense>
    class StringDictionaryColumnVisitor
            : public DictionaryColumnVisitor<int32_t, TFilter, ExtractValues, isDense> {
        using super = ColumnVisitor<int32_t, TFilter, ExtractValues, isDense>;
        using DictSuper =
        DictionaryColumnVisitor<int32_t, TFilter, ExtractValues, isDense>;

    public:
        StringDictionaryColumnVisitor(
            TFilter &filter,
            SelectiveColumnReader *reader,
            RowSet rows,
            ExtractValues values)
            : DictionaryColumnVisitor<int32_t, TFilter, ExtractValues, isDense>(
                filter,
                reader,
                rows,
                values) {
        }

        MELON_ALWAYS_INLINE vector_size_t process(int32_t value, bool &atEnd) {
            bool inStrideDict = !DictSuper::isInDict();
            auto index = value;
            if (inStrideDict) {
                index += DictSuper::dictionarySize();
            }
            vector_size_t previous =
                    isDense && TFilter::deterministic ? 0 : super::currentRow();
            if constexpr (!DictSuper::hasFilter()) {
                super::filterPassed(index);
            } else {
                // check the dictionary cache
                if (TFilter::deterministic &&
                    DictSuper::filterCache()[index] == FilterResult::kSuccess) {
                    super::filterPassed(index);
                } else if (
                    TFilter::deterministic &&
                    DictSuper::filterCache()[index] == FilterResult::kFailure) {
                    super::filterFailed();
                } else {
                    if (pollux::common::applyFilter(
                        super::filter_, valueInDictionary(index))) {
                        super::filterPassed(index);
                        if (TFilter::deterministic) {
                            DictSuper::filterCache()[index] = FilterResult::kSuccess;
                        }
                    } else {
                        super::filterFailed();
                        if (TFilter::deterministic) {
                            DictSuper::filterCache()[index] = FilterResult::kFailure;
                        }
                    }
                }
            }
            if (++super::rowIndex_ >= super::numRows_) {
                atEnd = true;
                return (TFilter::deterministic && isDense)
                           ? 0
                           : super::rows_[super::numRows_ - 1] - previous;
            }
            if (isDense && TFilter::deterministic) {
                return 0;
            }
            return super::currentRow() - previous - 1;
        }

        // Feeds'numValues' items starting at 'values' to the result. If
        // projecting out do nothing. If hook, call hook. If filter, apply
        // and produce hits and if not filter only compact the values to
        // remove non-passing. Returns the number of values in the result
        // after processing.
        template<bool hasFilter, bool hasHook, bool scatter>
        void processRun(
            const int32_t *input,
            int32_t numInput,
            const int32_t *scatterRows,
            int32_t *filterHits,
            int32_t *values,
            int32_t &numValues) {
            DKCHECK(input == values + numValues);
            setByInDict(values + numValues, numInput);
            if constexpr (!DictSuper::hasFilter()) {
                if (hasHook) {
                    for (auto i = 0; i < numInput; ++i) {
                        super::values_.addValue(
                            scatterRows
                                ? scatterRows[super::rowIndex_ + i]
                                : super::rowIndex_ + i,
                            valueInDictionary(input[i]));
                    }
                }
                if constexpr (std::is_same_v<TFilter, pollux::common::IsNotNull>) {
                    auto *begin = (scatter ? scatterRows : super::rows_) + super::rowIndex_;
                    std::copy(begin, begin + numInput, filterHits + numValues);
                    numValues += numInput;
                } else if constexpr (scatter) {
                    dwio::common::scatterDense(
                        input, scatterRows + super::rowIndex_, numInput, values);
                    numValues = scatterRows[super::rowIndex_ + numInput - 1] + 1;
                } else {
                    numValues += numInput;
                }
                super::rowIndex_ += numInput;
                return;
            } else {
                static_assert(hasFilter);
            }
            constexpr bool filterOnly =
                    std::is_same_v<typename super::Extract, DropValues>;
            constexpr int32_t kWidth = ksimd::batch<int32_t>::size;
            for (auto i = 0; i < numInput; i += kWidth) {
                auto indices = ksimd::load_unaligned(input + i);
                ksimd::batch<int32_t> cache;
                auto base =
                        reinterpret_cast<const int32_t *>(DictSuper::filterCache() - 3);
                if (i + kWidth > numInput) {
                    cache = simd::maskGather<int32_t, int32_t, 1>(
                        ksimd::broadcast<int32_t>(0),
                        simd::leadingMask<int32_t>(numInput - i),
                        base,
                        indices);
                } else {
                    cache = simd::gather<int32_t, int32_t, 1>(base, indices);
                }
                auto unknowns = simd::toBitMask(ksimd::batch_bool<int32_t>(
                    simd::reinterpretBatch<uint32_t>((cache & (kUnknown << 24)) << 1)));
                auto passed = simd::toBitMask(
                    ksimd::batch_bool<int32_t>(simd::reinterpretBatch<uint32_t>(cache)));
                if (UNLIKELY(unknowns)) {
                    uint16_t bits = unknowns;
                    while (bits) {
                        int index = bits::getAndClearLastSetBit(bits);
                        int32_t value = input[i + index];
                        if (applyFilter(super::filter_, valueInDictionary(value))) {
                            DictSuper::filterCache()[value] = FilterResult::kSuccess;
                            passed |= 1 << index;
                        } else {
                            DictSuper::filterCache()[value] = FilterResult::kFailure;
                        }
                    }
                }
                if (!passed) {
                    continue;
                } else if (passed == (1 << kWidth) - 1) {
                    ksimd::load_unaligned(
                                (scatter ? scatterRows : super::rows_) + super::rowIndex_ + i)
                            .store_unaligned(filterHits + numValues);
                    if (!filterOnly) {
                        indices.store_unaligned(values + numValues);
                    }
                    numValues += kWidth;
                } else {
                    int8_t numBits = __builtin_popcount(passed);
                    simd::filter(
                                ksimd::load_unaligned(
                                    (scatter ? scatterRows : super::rows_) + super::rowIndex_ + i),
                                passed)
                            .store_unaligned(filterHits + numValues);
                    if (!filterOnly) {
                        simd::filter(indices, passed).store_unaligned(values + numValues);
                    }
                    numValues += numBits;
                }
            }
            super::rowIndex_ += numInput;
        }

        // Processes a run length run.
        // 'value' is the value for 'currentRow' and numRows is the number of
        // selected rows that fall in this RLE. If value is 10 and delta is 3
        // and rows is {20, 30}, then this processes a 25 at 20 and a 40 at
        // 30.
        template<bool hasFilter, bool hasHook, bool scatter>
        void processRle(
            int32_t value,
            int32_t delta,
            int32_t numRows,
            int32_t currentRow,
            const int32_t *scatterRows,
            int32_t *filterHits,
            int32_t *values,
            int32_t &numValues) {
            constexpr int32_t kWidth = ksimd::batch<int32_t>::size;
            for (auto i = 0; i < numRows; i += kWidth) {
                ((ksimd::load_unaligned(super::rows_ + super::rowIndex_ + i) -
                  currentRow) *
                 delta +
                 value)
                        .store_unaligned(values + numValues + i);
            }

            processRun<hasFilter, hasHook, scatter>(
                values + numValues,
                numRows,
                scatterRows,
                filterHits,
                values,
                numValues);
        }

    private:
        void setByInDict(int32_t *values, int numValues) {
            if (DictSuper::inDict()) {
                auto current = super::rowIndex_;
                int32_t i = 0;
                for (; i < numValues; ++i) {
                    if (!bits::isBitSet(DictSuper::inDict(), super::rows_[i + current])) {
                        values[i] += DictSuper::dictionarySize();
                    }
                }
            }
        }

        melon::StringPiece valueInDictionary(int64_t index) {
            auto stripeDictSize = DictSuper::state_.dictionary.numValues;
            if (index < stripeDictSize) {
                return reinterpret_cast<const StringView *>(
                    DictSuper::state_.dictionary.values)[index];
            }
            return reinterpret_cast<const StringView *>(
                DictSuper::state_.dictionary2.values)[index - stripeDictSize];
        }
    };

    template<typename T, typename TFilter, typename ExtractValues, bool isDense>
    class DirectRleColumnVisitor
            : public ColumnVisitor<T, TFilter, ExtractValues, isDense> {
        using super = ColumnVisitor<T, TFilter, ExtractValues, isDense>;

    public:
        DirectRleColumnVisitor(
            TFilter &filter,
            SelectiveColumnReader *reader,
            RowSet rows,
            ExtractValues values)
            : ColumnVisitor<T, TFilter, ExtractValues, isDense>(
                filter,
                reader,
                rows,
                values) {
        }

        // Processes 'numInput' T's in 'input'. Sets 'values' and
        // 'numValues'' to the resulting values. 'scatterRows' may be
        // non-null if there is no filter and the decoded values should be
        // scattered into values with gaps in between so as to leave gaps
        // for nulls. If scatterRows is given, the ith value goes to
        // values[scatterRows[i]], else it goes to 'values[i]'. If
        // 'hasFilter' is true, the passing values are written to
        // consecutive places in 'values'.
        template<bool hasFilter, bool hasHook, bool scatter>
        void processRun(
            const T *input,
            int32_t numInput,
            const int32_t *scatterRows,
            int32_t *filterHits,
            T *values,
            int32_t &numValues) {
            DKCHECK_EQ(input, values + numValues);
            constexpr bool filterOnly =
                    std::is_same_v<typename super::Extract, DropValues>;

            dwio::common::processFixedWidthRun<T, filterOnly, scatter, isDense>(
                melon::Range<const vector_size_t *>(super::rows_, super::numRows_),
                super::rowIndex_,
                numInput,
                scatterRows,
                values,
                filterHits,
                numValues,
                super::filter_,
                super::values_.hook());

            super::rowIndex_ += numInput;
        }

        template<bool hasFilter, bool hasHook, bool scatter>
        void processRle(
            T value,
            T delta,
            int32_t numRows,
            int32_t currentRow,
            const int32_t *scatterRows,
            int32_t *filterHits,
            T *values,
            int32_t &numValues) {
            if (sizeof(T) == 8) {
                constexpr int32_t kWidth = ksimd::batch<int64_t>::size;
                for (auto i = 0; i < numRows; i += kWidth) {
                    auto numbers = detail::cvtU32toI64(
                                       simd::loadGatherIndices<int64_t>(
                                           super::rows_ + super::rowIndex_ + i) -
                                       currentRow) *
                                   delta +
                                   value;
                    numbers.store_unaligned(values + numValues + i);
                }
            } else if (sizeof(T) == 4) {
                constexpr int32_t kWidth = ksimd::batch<int32_t>::size;
                for (auto i = 0; i < numRows; i += kWidth) {
                    auto numbers =
                            (ksimd::load_unaligned(super::rows_ + super::rowIndex_ + i) -
                             currentRow) *
                            static_cast<int32_t>(delta) +
                            static_cast<int32_t>(value);
                    numbers.store_unaligned(values + numValues + i);
                }
            } else {
                for (auto i = 0; i < numRows; ++i) {
                    values[numValues + i] =
                            (super::rows_[super::rowIndex_ + i] - currentRow) * delta + value;
                }
            }

            processRun<hasFilter, hasHook, scatter>(
                values + numValues,
                numRows,
                scatterRows,
                filterHits,
                values,
                numValues);
        }
    };

    template<bool kEncodingHasNulls, bool kDictionary>
    class StringColumnReadWithVisitorHelper {
    public:
        StringColumnReadWithVisitorHelper(SelectiveColumnReader &reader, RowSet rows)
            : reader_(reader), rows_(rows) {
        }

        template<typename F>
        auto operator()(F &&readWithVisitor) {
            const bool isDense = rows_.back() == rows_.size() - 1;
            if (reader_.scanSpec()->keepValues()) {
                if (auto *hook = reader_.scanSpec()->valueHook()) {
                    if (isDense) {
                        readHelper<pollux::common::AlwaysTrue, true>(
                            &alwaysTrue(),
                            ExtractToGenericHook(hook),
                            std::forward<F>(readWithVisitor));
                    } else {
                        readHelper<pollux::common::AlwaysTrue, false>(
                            &alwaysTrue(),
                            ExtractToGenericHook(hook),
                            std::forward<F>(readWithVisitor));
                    }
                } else {
                    if (isDense) {
                        processFilter<true>(
                            ExtractToReader(&reader_), std::forward<F>(readWithVisitor));
                    } else {
                        processFilter<false>(
                            ExtractToReader(&reader_), std::forward<F>(readWithVisitor));
                    }
                }
            } else {
                if (isDense) {
                    processFilter<true>(DropValues(), std::forward<F>(readWithVisitor));
                } else {
                    processFilter<false>(DropValues(), std::forward<F>(readWithVisitor));
                }
            }
        }

    private:
        template<typename TFilter, bool kIsDense, typename ExtractValues, typename F>
        void readHelper(
            pollux::common::Filter *filter,
            ExtractValues extractValues,
            F readWithVisitor) {
            readWithVisitor(
                ColumnVisitor<melon::StringPiece, TFilter, ExtractValues, kIsDense>(
                    *static_cast<TFilter *>(filter), &reader_, rows_, extractValues));
        }

        template<bool kIsDense, typename ExtractValues, typename F>
        void processFilter(ExtractValues extractValues, F &&readWithVisitor) {
            using FilterValueT =
                    std::conditional_t<kDictionary, vector_size_t, StringView>;
            auto *filter = reader_.scanSpec()->filter();
            if (filter == nullptr) {
                readHelper<pollux::common::AlwaysTrue, kIsDense>(
                    &alwaysTrue(), extractValues, std::forward<F>(readWithVisitor));
                return;
            }
            switch (filter->kind()) {
                case pollux::common::FilterKind::kAlwaysTrue:
                    readHelper<pollux::common::AlwaysTrue, kIsDense>(
                        filter, extractValues, std::forward<F>(readWithVisitor));
                    break;
                case pollux::common::FilterKind::kIsNull:
                    if constexpr (kEncodingHasNulls) {
                        reader_.filterNulls<FilterValueT>(
                            rows_, true, !std::is_same_v<ExtractValues, DropValues>);
                    } else {
                        readHelper<pollux::common::IsNull, kIsDense>(
                            filter, extractValues, std::forward<F>(readWithVisitor));
                    }
                    break;
                case pollux::common::FilterKind::kIsNotNull:
                    if constexpr (
                        kEncodingHasNulls && std::is_same_v<ExtractValues, DropValues>) {
                        reader_.filterNulls<FilterValueT>(rows_, false, false);
                    } else {
                        readHelper<pollux::common::IsNotNull, kIsDense>(
                            filter, extractValues, std::forward<F>(readWithVisitor));
                    }
                    break;
                case pollux::common::FilterKind::kBytesRange:
                    readHelper<pollux::common::BytesRange, kIsDense>(
                        filter, extractValues, std::forward<F>(readWithVisitor));
                    break;
                case pollux::common::FilterKind::kNegatedBytesRange:
                    readHelper<pollux::common::NegatedBytesRange, kIsDense>(
                        filter, extractValues, std::forward<F>(readWithVisitor));
                    break;
                case pollux::common::FilterKind::kBytesValues:
                    readHelper<pollux::common::BytesValues, kIsDense>(
                        filter, extractValues, std::forward<F>(readWithVisitor));
                    break;
                case pollux::common::FilterKind::kNegatedBytesValues:
                    readHelper<pollux::common::NegatedBytesValues, kIsDense>(
                        filter, extractValues, std::forward<F>(readWithVisitor));
                    break;
                default:
                    readHelper<pollux::common::Filter, kIsDense>(
                        filter, extractValues, std::forward<F>(readWithVisitor));
                    break;
            }
        }

        SelectiveColumnReader &reader_;
        const RowSet rows_;
    };
} // namespace kumo::pollux::dwio::common
